/*
 * Copyright (c) 2020 Carnegie Lee
 * All rights reserved
 */
package com.geekc.kafka.config;

import com.geekc.kafka.constant.KafkaConst;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.*;

/**
 * Kafka配置类
 * @author Carnegie Lee
 */
@Configuration
public class KafkaMainConfig {

    @Bean
    public KafkaAdmin kafkaAdmin(KafkaProperties kafkaProperties) {
        KafkaAdmin kafkaAdmin = new KafkaAdmin(kafkaProperties.buildAdminProperties());
        //Broker不可用时影响正常业务需要
        kafkaAdmin.setFatalIfBrokerNotAvailable(true);
        return kafkaAdmin;
    }

    /**
     * 创建主题 - 非Replying模式
     * @return
     */
    //@Bean
    public NewTopic targetSimpleTopic() {
        /** 主题名称 */
        String topicName = KafkaConst.KAFKA_TARGET_TOPIC_SIMPLE;
        /** 分区数 */
        int partitions = 1;
        /** 副本数，不超过Broker数量 */
        short replicationFactor = 1;
        NewTopic topic = new NewTopic(topicName, partitions, replicationFactor);
        return topic;
    }

    /**
     * 创建主题 - Replying模式
     * @return
     */
    @Bean
    public NewTopic targetReplyingTopic() {
        /** 主题名称 */
        String topicName = KafkaConst.KAFKA_TARGET_TOPIC_REPLYING;
        /** 分区数 */
        int partitions = 1;
        /** 副本数，不超过Broker数量 */
        short replicationFactor = 1;
        NewTopic topic = new NewTopic(topicName, partitions, replicationFactor);
        return topic;
    }
}
